/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.sss.common.io.DBSet;
import org.sss.common.io.KeyFormatter;
import org.sss.mapreduce.DB;
import org.sss.mapreduce.Decoder;
import org.sss.mapreduce.Encodings;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SideData;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.datatype.Packable;

public class SideDataOnDemand<K extends Packable, V extends Packable> implements SideData<K, V> {
  private static Logger logger = SssServer.getLogger(SideDataOnDemand.class);

  private class BinKey {
    final StorageNode node;
    final byte[]      binkey;

    public BinKey(StorageNode node, byte[] binkey) {
      this.node   = node;
      this.binkey = binkey;
    }
  }

  private Map<K, List<V>>      valueDic  = new HashMap<K, List<V>>();
  private Map<K, List<BinKey>> binkeyDic = new HashMap<K, List<BinKey>>();
  private Class<V> valueClass;
  private Decoder valueDecoder;
  private DBSet dbSet;

  public static <KEY extends Packable, VALUE extends Packable> SideData<KEY, VALUE> create(
      SssServerResource sssResouce,
      StorageNode node,
      GroupID gid,
      Encodings encodings,
      Class<KEY> keyClass,
      Class<VALUE> valueClass) throws SssException {
    return new SideDataOnDemand<KEY, VALUE>(sssResouce, node, gid, encodings, keyClass, valueClass);
  }

  // TODO: uses appropriate exception.
  private SideDataOnDemand(SssServerResource sssResouce,
                  StorageNode nearNode,
                  GroupID gid,
                  Encodings encodings,
                  Class<K> keyClass,
                  Class<V> valueClass) throws SssException {
    try {
      this.dbSet = sssResouce.getDBSetForSideData();

      Map<StorageNode, List<byte[]>> results = new HashMap<StorageNode, List<byte[]>>();
      final byte[] prefix = KeyFormatter.getGroupIDPrefix(gid);

      if (gid.isBroadcast()) {
        results.put(nearNode, this.dbSet.get(nearNode).fwmkeys(prefix, -1));
      }
      else {
        for (StorageNode node: sssResouce.getStorageNodeList()) {
          results.put(node, this.dbSet.get(node).fwmkeys(prefix, -1));
        }
      }

      KeyFormatter kf = KeyFormatter.get(gid.getKeyFormat());
      Decoder dec = encodings.createDecoder(keyClass);

      // TODO: decodes parallell?
      // TODO: decodes in Read Thread.
      for (Map.Entry<StorageNode, List<byte[]>> e: results.entrySet()) {
        for (byte[] syskey: e.getValue()) {
          K userkey = keyClass.cast(dec.decode(kf.getUserKey(syskey)));
          List<BinKey> bkl = binkeyDic.get(userkey);
          if (bkl == null) {
            bkl = new ArrayList<BinKey>();
            binkeyDic.put(userkey, bkl);
          }

          bkl.add(new BinKey(e.getKey(), syskey));
        }
      }
      this.valueClass   = valueClass;
      this.valueDecoder = encodings.createDecoder(valueClass);
    }
    catch (IOException e) {
      throw new SssException(e);
    }
  }

  public boolean containsKey(K key) {
    return binkeyDic.containsKey(key);
  }

  // TODO: uses appropriate exception.
  public synchronized List<V> get(K key) throws SssException {
    try {
      List<V> vs = valueDic.get(key);
      if (vs != null) {
        return vs;
      }
      final List<BinKey> bks = binkeyDic.get(key);
      if (bks == null) {
        return null;
      }

      List<V> vals = new ArrayList<V>();
      for (BinKey bk: bks) {
        DB db = dbSet.get(bk.node);
        vals.add(valueClass.cast(valueDecoder.decode(db.get(bk.binkey))));
      }

      valueDic.put(key, vals);

      return vals;
    } catch (IOException e) {
      throw new SssException(e);
    }
  }

  public Iterable<K> keys() throws SssException {
    return binkeyDic.keySet();
  }

  public Iterable<Tuple<K, V>> tuples() throws SssException {
    return new Iterable<Tuple<K, V>>() {
      @Override
      public Iterator<Tuple<K, V>> iterator() {
        return new TupleIterator<K, V>(SideDataOnDemand.this);
      }
    };
  }

  @Override
  public synchronized void close() throws IOException {
    if (dbSet != null) {
      dbSet.close();
      dbSet = null;
    }
  }

  @Override
  public void finalize() {
    try {
      close();
    } catch (IOException e) {
      logger.error("Exception in finalize().", e);
    }
  }
}
